package com.spx.chapter09;

import com.spx.util.ClickSource;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * create by undeRdoG on  2022-05-08  15:21
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class CheckpointAndSavepointTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(10000L);

        env.setStateBackend(new HashMapStateBackend());

        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        checkpointConfig.setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://hadoop102:8020/flink/checkpoints"));

        // 设置checkpoint的超时时间
        checkpointConfig.setCheckpointTimeout(6000L);

        // 设置精确一次性语义
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        /*
        *  以下的配置均是在checkpoint的完成时间过长，而需要配置的配置项
        *  setMinPauseBetweenCheckpoints 的 优先级 应该是高于 setMaxConcurrentCheckpoints
        *  即只要配置了setMinPauseBetweenCheckpoints ，同一时刻下，最多应该只有一个checkpoint
        * */

        // 设置两次checkpoint之间最小的时间间隔
        checkpointConfig.setMinPauseBetweenCheckpoints(500L);

        // 同时checkpoint的最多个数
        //checkpointConfig.setMaxConcurrentCheckpoints(2);

        // 开启不对齐的checkpoint
        checkpointConfig.enableUnalignedCheckpoints();

        // 允许checkpoint失败的次数，默认是0，即不容忍，如果checkpoint失败，那么整个作业也就会失败
        checkpointConfig.setTolerableCheckpointFailureNumber(3);


        env.addSource(new ClickSource())
                .print();

        System.setProperty("HADOOP_USER_NAME", "shipeixin");


        env.execute();
    }
}
